package learn_concurrent._14;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class T1 {
	static final Logger logger = LoggerFactory.getLogger(T1.class);
	public static void main(String[] args) {
		ThreadPool pool = new ThreadPool(2, 1000, TimeUnit.MILLISECONDS, 10);
		for (int i = 0; i < 5; i++) {
			int j = i;
			pool.execute(() -> logger.debug("{}", j));
		}
	}

}

class ThreadPool {
	private static final Logger logger = LoggerFactory.getLogger(ThreadPool.class);
	
	private BlockingQueue<Runnable> taskQueue;
	private HashSet<Worker> workers = new HashSet<>();
	private int coreSize;
	private long timeout;
	private TimeUnit timeUnit;

	public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit, int queueCapacity) {
		this.coreSize = coreSize;
		this.timeout = timeout;
		this.timeUnit = timeUnit;
		this.taskQueue = new BlockingQueue<>(queueCapacity);
	}

	public void execute(Runnable task) {
		synchronized (workers) {
			if (workers.size() < coreSize) {
				Worker w = new Worker(task);
				logger.debug("Add a worker with a task, {}, {}", w, task);
				workers.add(w);
				w.start();
			} else {
				logger.debug("Put a task to queue, {}", task);
				taskQueue.put(task);
			}
		}
	}

	class Worker extends Thread {
		private final Logger logger = LoggerFactory.getLogger(Worker.class);
		private Runnable task;

		public Worker(Runnable task) {
			this.task = task;
		}

		@Override
		public void run() {
//			while (task != null || (task = taskQueue.take()) != null) {
			while (task != null || (task = taskQueue.poll(timeout, timeUnit)) != null) {
				try {
					logger.debug("Task run, {}", task);
					task.run();
				} catch (Exception e) {
					e.printStackTrace();
				} finally {
					task = null;
				}
			}
			synchronized (workers) {
				logger.debug("Remove worker, {}", this);
				workers.remove(this);
			}
		}
	}

}

class BlockingQueue<T> {
	private final static Logger logger = LoggerFactory.getLogger(BlockingQueue.class);
	
	private Deque<T> queue = new ArrayDeque<>();
	private ReentrantLock lock = new ReentrantLock();
	private Condition fullWaitSet = lock.newCondition();
	private Condition emptyWaitSet = lock.newCondition();
	private int capacity;

	public BlockingQueue(int capacity) {
		this.capacity = capacity;
	}

	public T poll(long timeout, TimeUnit unit) {
		lock.lock();
		try {
			long nanos = unit.toNanos(timeout);
			while (queue.isEmpty()) {
				try {
					if (nanos <= 0) {
						return null;
					}
					nanos = emptyWaitSet.awaitNanos(nanos);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			T t = queue.removeFirst();
			fullWaitSet.signal();
			return t;
		} finally {
			lock.unlock();
		}
	}

	public T take() {
		lock.lock();
		try {
			while (queue.isEmpty()) {
				try {
					emptyWaitSet.await();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			T t = queue.removeFirst();
			fullWaitSet.signal();
			return t;
		} finally {
			lock.unlock();
		}
	}

	public void put(T t) {
		lock.lock();
		try {
			while (queue.size() == capacity) {
				try {
					fullWaitSet.await();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			queue.addLast(t);
			emptyWaitSet.signal();
		} finally {
			lock.unlock();
		}
	}

	public int size() {
		lock.lock();
		try {
			return queue.size();
		} finally {
			lock.unlock();
		}
	}
}